package com.atguigu.flink.watermark;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/3/1
 * 
 *
 *  可以为一个流中的每个元素指定 一个时间间隔范围 [上边界,下边界]，只要在这个时间范围中到达的对方流中的数据
 *  都可以关联。打破了windowjoin中必须进入同一个窗口的限制。
 *
 *  必须是keyedStream才可以。
 *
 */
public class Demo9_IntervalJoin
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner( (e, ts) -> e.getTs());

        env.setParallelism(1);

        KeyedStream<WaterSensor, String> ds1 = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy)
            .keyBy(WaterSensor::getId);

        KeyedStream<WaterSensor, String> ds2 = env
            .socketTextStream("hadoop103", 8889)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy)
            .keyBy(WaterSensor::getId);


        /*
                老版本支持inProcessingTime，新的版本都不支持。写上就报错。

                    左边流 join 右边流。
                        根据木桶理论，如果右侧流中数据的 eventtime < 下游process的水印，称为迟到数据，迟到无法 关联。
         */
        ds1.intervalJoin(ds2)
           //.inEventTime()
           //指定时间范围边界 [-3,3]
           .between(Time.seconds(-3),Time.seconds(3))
           .process(new ProcessJoinFunction<WaterSensor, WaterSensor, String>()
           {
               // 没有窗口的话，process是一个元素调用一次。 在这个场景下，就是Join成功，就调用一次。
               @Override
               public void processElement(WaterSensor left, WaterSensor right, Context ctx, Collector<String> out) throws Exception {
                        out.collect(left + "====" +right);
               }
           })
           //刚好卡在边界是否join，默认是Join
           /*.lowerBoundExclusive()
           .upperBoundExclusive()*/
           .print();




        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
        
    }
    
}
